package com.imooc.bigdata.hos.server.filemgr.service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Service;

import com.imooc.bigdata.hos.core.ErrorCodes;
import com.imooc.bigdata.hos.server.HosServerException;

/**
 * Created by jixin on 17-2-22.
 */
@Service("hBaseServiceImpl")
public class HBaseServiceImpl implements IHBaseService {

  @Override
  public boolean createTable(Connection connection, String tableName, String[] cfs, byte[][] splitKeys) {
    try (HBaseAdmin admin = (HBaseAdmin) connection.getAdmin()) {

      if (admin.tableExists(tableName)) {
        return false;
      }

      HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));
      for (int i = 0; i < cfs.length; i++) {
        HColumnDescriptor hcolumnDesc = new HColumnDescriptor(cfs[i]);
        hcolumnDesc.setMaxVersions(1);
        tableDesc.addFamily(hcolumnDesc);
      }

      //创建预先分区的表
      admin.createTable(tableDesc, splitKeys);

    } catch (Exception e) {
      String msg = String.format("create table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }

    return true;
  }

  @Override
  public  boolean createTable(Connection connection, String tableName, String[] cfs) {
//    //todo 可以写成这样吗
//    createTable(connection,tableName,cfs,null);

    try (HBaseAdmin admin = (HBaseAdmin) connection.getAdmin()) {

      if (admin.tableExists(tableName)) {
        return false;
      }

      HTableDescriptor tableDesc = new HTableDescriptor(TableName.valueOf(tableName));
      for (int i = 0; i < cfs.length; i++) {
        HColumnDescriptor hcolumnDescriptor = new HColumnDescriptor(cfs[i]);
        hcolumnDescriptor.setMaxVersions(1);
        tableDesc.addFamily(hcolumnDescriptor);
      }
      admin.createTable(tableDesc);
    } catch (Exception e) {
      String msg = String.format("create table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }

    return true;
  }

  @Override
  public  boolean deleteTable(Connection connection, String tableName) {
    try (HBaseAdmin admin = (HBaseAdmin) connection.getAdmin()) {

      //删除表之前需要先disable
      admin.disableTable(tableName);
      admin.deleteTable(tableName);

    } catch (Exception e) {
      String msg = String.format("delete table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }

    return true;
  }

  @Override
  public  boolean deleteColumnFamily(Connection connection,
                                     String tableName,
                                     String columnFamilyName) {
    try (HBaseAdmin admin = (HBaseAdmin) connection.getAdmin()) {
      admin.deleteColumn(tableName, columnFamilyName);
    } catch (IOException e) {
      String msg = String.format("delete table=%s , column family=%s error. msg=%s",
                                tableName,
                                columnFamilyName,
                                e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
    return true;
  }

  @Override
  public  boolean deleteColumnQualifier(Connection connection,
                                        String tableName,
                                        String rowKey,
                                        String columnFamilyName,
                                        String qualifierName) {

      Delete delete = new Delete(rowKey.getBytes());
      delete.addColumns(columnFamilyName.getBytes(), qualifierName.getBytes());

      return deleteColumnQualifier(connection,tableName,delete);
  }

  @Override
  public  boolean deleteColumnQualifier(Connection connection,
                                        String tableName,
                                        Delete delete) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      table.delete(delete);
    } catch (IOException e) {
      String msg = String .format("delete table=%s , column family=%s , qualifier=%s error. msg=%s",
                                  tableName,
                                  e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }

    return true;
  }


  @Override
  public  boolean deleteRow(Connection connection,
                            String tableName,
                            String rowKey,
                            String columnFamilyName,
                            String qualifierName) {

    Delete delete = new Delete(rowKey.getBytes());
    delete.addColumns(columnFamilyName.getBytes(), qualifierName.getBytes());

    return deleteRow(connection,tableName,delete);
  }

  @Override
  public  boolean deleteRow(Connection connection,
                            String tableName,
                            Delete delete) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      table.delete(delete);
    } catch (IOException e) {
      String msg = String .format("delete table=%s , column family=%s , qualifier=%s error. msg=%s",
                                  tableName,
                                  e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }

    return true;
  }

  @Override
  public  boolean deleteRow(Connection connection,
                            String tableName,
                            String rowKey) {
      Delete delete = new Delete(Bytes.toBytes(rowKey));
      return deleteRow(connection,tableName,delete);
  }

  @Override
  public  boolean deleteRows(Connection connection,
                             String tableName,
                             List<String> rows) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      List<Delete> list = new ArrayList<Delete>();
      for (String row : rows) {
        Delete d = new Delete(Bytes.toBytes(row));
        list.add(d);
      }
      if (list.size() > 0) {
        table.delete(list);
      }
    } catch (IOException e) {
      String msg = String.format("delete table=%s , rows error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
    return true;
  }

  @Override
  public  boolean existsRow(Connection connection, String tableName, String row) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      Get g = new Get(Bytes.toBytes(row));
      return table.exists(g);
    } catch (IOException e) {
      String msg = String
          .format("check exists row from table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
  }

  @Override
  public  Result getRow(Connection connection,
                        String tableName,
                        String row,
                        FilterList filterList) {
    Result rs;
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      Get g = new Get(Bytes.toBytes(row));
      g.setFilter(filterList);
      rs = table.get(g);
    } catch (IOException e) {
      String msg = String
          .format("get row from table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
    return rs;
  }

  @Override
  public  Result getRow(Connection connection,
                        String tableName,
                        Get get) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      return table.get(get);
    } catch (IOException e) {
      String msg = String.format("get row from table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
  }

  @Override
  public  Result getRow(Connection connection,
                        String tableName,
                        String rowKey) {
      Get get = new Get(Bytes.toBytes(rowKey));
      return getRow(connection,tableName,get);

  }

  @Override
  public  Result getRow(Connection connection,
                        String tableName,
                        String row,
                        byte[] column,
                        byte[] qualifier) {
    Result rs;
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      Get g = new Get(Bytes.toBytes(row));
      g.addColumn(column, qualifier);
      rs = table.get(g);
    } catch (IOException e) {
      String msg = String
          .format("get row from table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
    return rs;
  }

  @Override
  public  Result[] getRows(Connection connection,
                           String tableName,
                           List<String> rows,
                           FilterList filterList) {
    Result[] results = null;
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      List<Get> gets = null;
      gets = new ArrayList<Get>();
      for (String row : rows) {
        if (row != null) {
          Get g = new Get(Bytes.toBytes(row));
          g.setFilter(filterList);
          gets.add(g);
        }
      }
      if (gets.size() > 0) {
        results = table.get(gets);
      }
    } catch (IOException e) {
      String msg = String
          .format("get rows from table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
    return results;
  }


  @Override
  public  Result[] getRows(Connection connection,
                           String tableName,
                           List<String> rows) {
    Result[] results = null;
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      List<Get> gets = null;
      gets = new ArrayList<Get>();
      for (String row : rows) {
        if (row != null) {
          Get g = new Get(Bytes.toBytes(row));
          gets.add(g);
        }
      }
      if (gets.size() > 0) {
        results = table.get(gets);
      }
    } catch (IOException e) {
      String msg = String
          .format("get rows from table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
    return results;
  }


  @Override
  public  long incrementColumnValue(Connection connection,
                                    String tableName,
                                    String row,
                                    byte[] columnFamily,
                                    byte[] qualifier,
                                    int num) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      return table.incrementColumnValue(Bytes.toBytes(row), columnFamily, qualifier, num);
    } catch (IOException e) {
      String msg = String.format("incrementColumnValue table=%s error. msg=%s",
              tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
  }

  @Override
  public  boolean putRow(Connection connection,
                         String tableName,
                         String row,
                         String columnFamily,
                         String qualifier,
                         String data) {
    try {
      Put put = new Put(Bytes.toBytes(row));
      put.addColumn(Bytes.toBytes(columnFamily),
                    Bytes.toBytes(qualifier),
                    Bytes.toBytes(data));
      putRow(connection, tableName, put);
    } catch (Exception e) {
      String msg = String
          .format("put row from table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
    return true;
  }

  //  public  boolean putRow(Connection connection, String tableName, Put put) {
//    try {
//      putRows(connection, tableName, Arrays.asList(put));
//    } catch (Exception e) {
//      String msg = String.format("put row from table=%s error. msg=%s", tableName, e.getMessage());
//      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
//    }
//    return true;
//  }

  @Override
  public  boolean putRow(Connection connection, String tableName, Put put) {
    try(Table table = connection.getTable(TableName.valueOf(tableName))) {
      table.put(put);
    } catch (Exception e) {
      String msg = String.format("put row from table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
    return true;
  }

  @Override
  public  boolean putRow(Connection connection, String tableName, List<Put> puts) {
    try(Table table = connection.getTable(TableName.valueOf(tableName))) {
      table.put(puts);
    } catch (Exception e) {
      String msg = String.format("put row from table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
    return true;
  }

//  /**
//   * putRows.
//   *
//   * @param tableName tableName
//   * @param puts puts
//   * @return success of failed
//   */
//  public  boolean putRows(Connection connection, String tableName, List<Put> puts) {
//    long currentTime = System.currentTimeMillis();
//    final BufferedMutator.ExceptionListener listener = new BufferedMutator.ExceptionListener() {
//
//      public void onException(RetriesExhaustedWithDetailsException e, BufferedMutator mutator) {
//        String msg = String.format("put rows from table=%s error. msg=%s", tableName, e.getMessage());
//        throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
//      }
//    };
//    BufferedMutatorParams params = new BufferedMutatorParams(TableName.valueOf(tableName))
//                                                              .listener(listener);
//    params.writeBufferSize(5 * 1024 * 1024);
//    try (final BufferedMutator mutator = connection.getBufferedMutator(params)) {
//      mutator.mutate(puts);
//      mutator.flush();
//    } catch (IOException e) {
//      String msg = String.format("put rows from table=%s error. msg=%s", tableName, e.getMessage());
//      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
//    }
//    return true;
//  }

  @Override
  public  ResultScanner scanner(Connection connection, String tableName) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      Scan scan = new Scan();
      scan.setCaching(1000);
      return table.getScanner(scan);
    } catch (IOException e) {
      String msg = String.format("scan table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
  }

  @Override
  public  ResultScanner scanner(Connection connection,
                                String tableName,
                                String startRowKey,
                                String stopRowKey) {
    return scanner(connection, tableName, Bytes.toBytes(startRowKey), Bytes.toBytes(stopRowKey));
  }

  @Override
  public  ResultScanner scanner(Connection connection,
                                String tableName,
                                byte[] startRowKey,
                                byte[] stopRowKey) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      Scan scan = new Scan();
      scan.setStartRow(startRowKey);
      scan.setStopRow(stopRowKey);
      scan.setCaching(1000);
      return table.getScanner(scan);
    } catch (IOException e) {
      String msg = String.format("scan table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
  }

  @Override
  public  ResultScanner scanner(Connection connection,
                                String tableName,
                                FilterList filterList) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      Scan scan = new Scan();
      scan.setCaching(1000);
      scan.setFilter(filterList);
      return table.getScanner(scan);
    } catch (IOException e) {
      String msg = String
              .format("scan table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
  }

  @Override
  public  ResultScanner scanner(Connection connection,
                                String tableName,
                                Scan scan) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      return table.getScanner(scan);
    } catch (IOException e) {
      String msg = String
              .format("scan table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
  }

  @Override
  public  ResultScanner scanner(Connection connection,
                                String tableName,
                                byte[] startRowKey,
                                byte[] stopRowKey,
                                FilterList filterList) {
    try (Table table = connection.getTable(TableName.valueOf(tableName))) {
      Scan scan = new Scan();
      scan.setStartRow(startRowKey);
      scan.setStopRow(stopRowKey);
      scan.setCaching(1000);
      scan.setFilter(filterList);
      return table.getScanner(scan);
    } catch (IOException e) {
      String msg = String
              .format("scan table=%s error. msg=%s", tableName, e.getMessage());
      throw new HosServerException(ErrorCodes.ERROR_HBASE, msg);
    }
  }

  @Override
  public  ResultScanner scanner(Connection connection,
                                String tableName,
                                String startRowKey,
                                String stopRowKey,
                                FilterList filterList) {
    return scanner(connection, tableName, Bytes.toBytes(startRowKey), Bytes.toBytes(stopRowKey),filterList);
  }
}
